package com.example.reactstream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * publisher->processor->subscriber 的关系可以参考/pic/publisher-processor-subscriber.png 这张图
 */
public class Reactstreamdemo02 {

    public static void main(String[] args) throws InterruptedException {

        //实例化发布者
        SubmissionPublisher<String> publisher=new SubmissionPublisher<>();

        //实例化处理器
        ReactProcessor reactProcessor=new ReactProcessor();

        //发布者建立与处理器的关系
        publisher.subscribe(reactProcessor);

        //创建订阅者
        Flow.Subscriber subscriber=new Flow.Subscriber() {

            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription=subscription;
                System.out.println("建立processor 和 subscriber的关系");
                subscription.request(1);
            }

            @Override
            public void onNext(Object o) {
                System.out.println("接收数据："+o);

                //订阅者处理完数据还需要接着处理多少条进行处理  这就是背压，也就是订阅者处理多少条数据不是由发布者控制的，而是由这个背压控制的
                subscription.request(1);

            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("订阅者异常");

            }

            @Override
            public void onComplete() {
                System.out.println("订阅者处理完数据");

            }
        };

        //处理器建立与订阅者的关系
        reactProcessor.subscribe(subscriber);

        for (int i = 0; i < 300; i++) {
            System.out.println("发布者发布数据:"+"publisher-processor-subscriber: "+i);
            publisher.submit("publisher-processor-subscriber: "+i);
        }

        //关闭发布者
        publisher.close();


        Thread.currentThread().join();
    }
}
